package org.tao.lightningmq.common.remote;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.common.tcp.TcpMsgDecoder;
import org.tao.lightningmq.common.tcp.TcpMsgEncoder;

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
 * @Author lht
 * @date 2025/1/8 - 17:27
 * @description:
 */
public class NameServerNettyRemoteClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(NameServerNettyRemoteClient.class);
    private static final int MAX_RETRY_TIMES = 20;

    private final String ip;
    private final int port;
    private final Bootstrap bootstrap;
    private Channel channel;
    private NameServerRemoteRespHandler handler;
    private BrokerMasterChangeCallBack brokerMasterChangeCallBack;
    private NameserverLogoutCallBack nameServerLogoutCallBack;

    public NameServerNettyRemoteClient(String ip, int port) {
        this.ip = ip;
        this.port = port;
        EventLoopGroup clientGroup = new NioEventLoopGroup(1);
        bootstrap = new Bootstrap();
        bootstrap.group(clientGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new TcpMsgDecoder());
                ch.pipeline().addLast(new TcpMsgEncoder());
                handler = new NameServerRemoteRespHandler();
                handler.setNameServerLogoutCallBack(nameServerLogoutCallBack);
                handler.setBrokerMasterChangeCallBack(brokerMasterChangeCallBack);
                ch.pipeline().addLast(handler);
            }
        });
        Runtime.getRuntime().addShutdownHook(new Thread(clientGroup::shutdownGracefully));
    }

    public boolean buildConnection() {
        return buildConnectionSync(0);

    }

    private boolean buildConnectionSync(int retryTimes) {
        if (retryTimes >= MAX_RETRY_TIMES) {
            LOGGER.warn("已达到最大连接重试次数!");
            return false;
        }
        try {
            LOGGER.info("开始和ns建立连接");
            ChannelFuture channelFuture = bootstrap.connect(ip, port)
                    .addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture channelFuture) throws Exception {
                            if (!channelFuture.isSuccess()) {
                                LOGGER.warn("与Nameserver连接未能成功，尝试进行重连!");
                            }
                        }
                    });
            channelFuture.await();
            if (channelFuture.isSuccess()) {
                channel = channelFuture.channel();
                return true;
            }
            // 连接失败，睡5秒后进行重连
            TimeUnit.SECONDS.sleep(5);
            return buildConnectionSync(retryTimes + 1);
        } catch (InterruptedException e) {
            LOGGER.error("连接过程中出现异常: ", e);
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
            }
            return buildConnectionSync(retryTimes + 1);
        }
    }

    public TcpMsg sendSyncMsg(TcpMsg tcpMsg, String msgId)  {
        SyncFuture syncFuture = new SyncFuture();
        syncFuture.setMsgId(msgId);
        SyncFutureManager.put(msgId, syncFuture);
        channel.writeAndFlush(tcpMsg);
        try {
            //return syncFuture.get(2, TimeUnit.SECONDS);
            return syncFuture.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void sendAsyncMsg(TcpMsg tcpMsg) {
        channel.writeAndFlush(tcpMsg);
    }

    public void setBrokerMasterChangeCallBack(BrokerMasterChangeCallBack brokerMasterChangeCallBack) {
        handler.setBrokerMasterChangeCallBack(brokerMasterChangeCallBack);
        this.brokerMasterChangeCallBack = brokerMasterChangeCallBack;
    }

    public void setNameServerLogoutCallBack(NameserverLogoutCallBack nameServerLogoutCallBack) {
        handler.setNameServerLogoutCallBack(nameServerLogoutCallBack);
        this.nameServerLogoutCallBack = nameServerLogoutCallBack;
    }

    public boolean isActive() {
        return this.channel.isActive();
    }

}
